package com.atbeijing.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atbeijing.gmall.realtime.bean.TableProcess;
import com.atbeijing.gmall.realtime.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class MyProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private OutputTag<JSONObject> dimTag;
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private Connection conn;

    public MyProcessFunction(OutputTag<JSONObject> dimTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.dimTag = dimTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //注册phoenix驱动
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        //建立phoenix连接
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    //处理主流,业务数据流
    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //获取主流中表的表名
        String tableName = value.getString("table");
        //获取主流中表的操作类型
        String type = value.getString("type");
        //注意：如果使用maxwell采集历史数据，那么操作类型是bootstrap-insert，需要改为insert
        if ("bootstrap-insert".equals(type)) {
            type = "insert";
            value.put("type", type);
        }
        //根据广播流拼接key的规则去拼接查询的key
        String key = tableName + ":" + type;
        //获取主流中表的配置信息
        TableProcess tableProcess = broadcastState.get(key);
        if (tableProcess!=null){
            //主流中的表应该放到哪个主题
            String sinkTable = tableProcess.getSinkTable();
            value.put("sink_table",sinkTable);
            //业务表data数据
            JSONObject data = value.getJSONObject("data");
            //配置表中该表需要的字段
            String sinkColumns = tableProcess.getSinkColumns();
            //根据配置表信息,对主流中业务表字段进行过滤
            if (sinkColumns!=null && sinkColumns.length()>0){
                //传递的是内存地址
                filterColumn(data,sinkColumns);
            }
            //是维度表还是事实表
            String sinkType = tableProcess.getSinkType();
            if (sinkType.equals(TableProcess.SINK_TYPE_HBASE)) {
                //维度数据 --发送到维度侧输出流中
                ctx.output(dimTag, value);
            } else if (sinkType.equals(TableProcess.SINK_TYPE_KAFKA)) {
                //事实数据---发送到主流
                out.collect(value);
            }
        }else {
            //没有找到配置信息
            System.out.println(">>>>>>业务表没有获取到配置表信息,key为: " + key);
        }

    }

    //处理广播流,数据库配置流
    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        //将读取到FlinkCDC采集到的信息  由jsonStr->jsonObj
        // {"database":"gmall1116","table":"comment_info","type":"insert","data":{}}
        JSONObject jsonObj = JSON.parseObject(value);
        //获取data数据  也就是每一条配置信息
        String dataJsonStr = jsonObj.getString("data");
        //将读取到的配置信息转换为TableProcess对象
        TableProcess tableProcess = JSON.parseObject(dataJsonStr, TableProcess.class);

        //获取源表表名
        String sourceTable = tableProcess.getSourceTable();
        //获取操作类型
        String operateType = tableProcess.getOperateType();
        //输出类型      hbase|kafka
        String sinkType = tableProcess.getSinkType();
        //输出目的地表名或者主题名
        String sinkTable = tableProcess.getSinkTable();
        //输出字段
        String sinkColumns = tableProcess.getSinkColumns();
        //表的主键
        String sinkPk = tableProcess.getSinkPk();
        //建表扩展语句
        String sinkExtend = tableProcess.getSinkExtend();
        //拼接key
        String key = sourceTable + ":" + operateType;

        //如果是新增的维度类型配置通过phoenix创建维度表
        if (sinkType.equals(TableProcess.SINK_TYPE_HBASE)&&"insert".equals(operateType)){
            //通过Phoenix创建表
            checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);
        }

        //将配置表信息放入广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        broadcastState.put(key,tableProcess);
    }


    //创建维度表,表名,字段,主键,扩展字段
    private void checkTable(String tableName, String fieldStr, String pk, String ext) {
        if (ext == null) {
            ext = "";
        }
        if (pk == null) {
            pk = "id";
        }
        StringBuilder createSql = new StringBuilder("create table if not exists "+ GmallConfig.HBASE_SCHEMA +"."+tableName+"(");
        //字段分割id,name,region_id,area_code,iso_code,iso_3166_2
        String[] fieldArr = fieldStr.split(",");
        for (int i = 0; i < fieldArr.length; i++) {
            String field = fieldArr[i];
            if(pk.equals(field)){
                createSql.append(field + " varchar primary key ");
            }else{
                createSql.append("info." + field + " varchar ");
            }
            //添加逗号
            if(i < fieldArr.length - 1){
                createSql.append(",");
            }
        }
        createSql.append(")" + ext);

        System.out.println("<<<<Phoenix维度表建表语句:" + createSql.toString());
        //创建表
        PreparedStatement ps = null;
        try {
            //创建数据库操作对象
            ps = conn.prepareStatement(createSql.toString());
            //执行SQL语句
            ps.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            //释放资源
            if(ps != null){
                try {
                    ps.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }



    //对字段进行过滤
    private void filterColumn(JSONObject data, String sinkColumns) {
        //对sinkColumns进行分割，得到是保留的字段的名称属性
        String[] fileds = sinkColumns.split(",");
        List<String> filedList = Arrays.asList(fileds);
        //json转型 set集合中装的kv对
        Set<Map.Entry<String, Object>> entrySet = data.entrySet();
        //删除此集合中满足给定条件的所有元素,r表示 业务表数据转的kv对
        entrySet.removeIf(r -> !filedList.contains(r.getKey()));
    }



}
